package com.htiiot.action;

import com.alibaba.fastjson.JSONObject;
import com.htiiot.base.UploadFile;
import com.htiiot.bean.LogBean;
import com.htiiot.config.Configs;
import com.htiiot.tools.Cmd;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.log4j.Logger;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.*;

public class Kafka2hdfs {
    private static Logger log = Logger.getLogger(Kafka2hdfs.class);
    public static void main(String[] args) throws Exception {
        Cmd.parseCmd(args);
        log.info("\r\n"+Configs.getInfo());
        Properties props = new Properties();
        props.put("bootstrap.servers", Configs.KafkaConfig.BOOTSTRAP);
        props.put("group.id", Configs.KafkaConfig.GROUPID);
        props.put("enable.auto.commit", "true");
        props.put("session.timeout.ms", "30000");
        props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        String rootDir = Configs.FileConfig.ROOT_DIR;
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        List<String> l = new ArrayList<String>();
        l.add(Configs.KafkaConfig.TOPIC);
        consumer.subscribe(l);
        FileWriter fwToday = null;
        FileWriter fwYes = null;
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
        Date dateToday = new Date();
        Date start = new Date(sdf.parse(sdf.format(dateToday)).getTime()+10*60*1000);
        log.info("everydayTime: "+new SimpleDateFormat("HH:mm:ss").format(start)+" going to update to HDFS");
        new Timer("UploadFile",true).scheduleAtFixedRate(new UploadFile(rootDir),start,24*60*60*1000);
        try {
            String fileName = "";
            long todayZero = 0l;
            while (true) {
                String today = sdf.format(new Date());
                if(!today.equals(fileName)){
                    if(fwToday!=null){
                        fwToday.close();
                    }
                    fileName = rootDir + "/" + today +"/log.txt";
                    createFile(fileName);
                    fwToday = new FileWriter(new File(fileName),true);
                    if(fwYes!=null){
                        delayClose(fwYes);
                    }
                    todayZero = sdf.parse(today).getTime();
                }
                ConsumerRecords<String, String> records = consumer.poll(1000);
                for(ConsumerRecord<String,String> r:records ){
                    String value = new String(r.value().getBytes(),"utf-8");
                    LogBean b = JSONObject.parseObject(value, LogBean.class);
                    if(b.getCreateTime()!=0) {
                        if (b.getCreateTime() < todayZero) {
                            log.info("This record create time before today\r\n"+value);
                            if (fwYes != null&&fwYes.getEncoding() != null) {
                                fwYes.write(b.format());
                                delayClose(fwYes);
                            } else {
                                String fileYes = rootDir + "/" + sdf.format(new Date(dateToday.getTime() - 86400000)) + "/log.txt";
                                createFile(fileYes);
                                fwYes = new FileWriter(fileYes, true);
                                fwYes.write(b.format());
                                delayClose(fwYes);
                            }
                        } else {
                            fwToday.write(b.format());
                        }
                    }else{
                        log.info("This record is old format auto ignore\r\n"+value);
                    }
                }
            }
        }catch (Exception e){
            e.printStackTrace();
            log.error(e,e);
        }finally {
            if(consumer!=null){
                consumer.close();
            }
            if(fwToday!=null) {
                fwToday.close();
            }
            if(fwYes!=null){
                fwYes.close();
            }
        }
    }

    public static void createFile(String file) throws IOException {
        File f = new File(file);
        if(!f.exists()){
            f.getParentFile().mkdirs();
            f.createNewFile();
        }
    }

    /**
     * 延迟关流
     */
    public static void delayClose( FileWriter fos){
        new Timer().schedule(new TimerTask() {
            @Override
            public void run() {
                try {
                    fos.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        },5*60*1000);
    }
}
